package com.qudian.pay.mq.producer.impl;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.alibaba.rocketmq.client.exception.MQBrokerException;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.SendCallback;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.remoting.exception.RemotingException;
import com.qudian.pay.core.utils.JdkSerializer;
import com.qudian.pay.mq.MQException;
import com.qudian.pay.mq.SendMessage;
import com.qudian.pay.mq.DefaultSendResult;
import com.qudian.pay.mq.enums.MQRuntimeFailStatus;
import com.qudian.pay.mq.producer.MQProducer;

/**
 * @Description: MQ生产者实现
 * @author gaoxiang
 * @date 2016年6月3日 上午10:48:04
 */ 
public class MQProducerImpl implements MQProducer {
	
	private final static Logger logger = LoggerFactory.getLogger(DefaultMQProducer.class);
	
	/**
	 * @Fields namesrvAddr : nameServer IP:Port
	 */ 
	private String namesrvAddr;
	
	/**
	 * @Fields instanceName : 同一个server上启动多个producer的标识信息
	 */ 
	private String instanceName;
	
	private String group;
	
	/**
	 * @Fields topicsACL : 允许发送的topics,多个使用逗号隔开
	 */ 
	private String topicsACL;
	
	/**
	 * @Fields tagsACL : 允许发送的tags,多个使用逗号隔开
	 */ 
	private String tagsACL;
	
	/**
	 * @Fields needCheckTopics : 是否对可以发送的topic进行检查的标示
	 */ 
	private boolean needCheckTopics = true;
	
	/**
	 * 是否对可以发送的tag进行检查的标示
	 */
	private boolean needCheckTags = true;
	
	private DefaultMQProducer producer;
	
	private int retryTimesWhenSendFailed;
	
	public void init() {
		checkPropertiesAndSetEnv();
		producer = new DefaultMQProducer();
		producer.setNamesrvAddr(namesrvAddr);
		producer.setInstanceName(instanceName);
		producer.setProducerGroup(group);
		// 修复mq超时重复发送的bug，将重试次数设置为0
		producer.setRetryTimesWhenSendFailed(retryTimesWhenSendFailed);
		try {
			producer.start();
			logger.info("RocketMQ生产者启动成功:nameserver={" + namesrvAddr + "},instanceName={" + instanceName + "},group={" + group + "}");
		} catch (MQClientException e) {
			logger.error("RocketMQ生产者启动失败:nameserver={" + namesrvAddr + "},instanceName={" + instanceName + "},group={" + group + "}", e);
		}
	}
	
	@Override
	public DefaultSendResult sendMessage(String topic, String tag, SendMessage msg) {
		if (doCheckTopic(topic) || doCheckTag(tag)) {
			throw new MQException(MQRuntimeFailStatus.MQ_PUBLISH_PERMISSION_FAILURE);
		}
		msg.setInstanceName(instanceName);
		byte[] body = null;
		try {
			body = JdkSerializer.serialize(msg);
		} catch (Exception e) {
			logger.error("序列化SendMessage信息异常", e);
		}
		Message mqMsg = new Message(topic, tag, body);
		if (msg.getDelayLevel() > 0) mqMsg.setDelayTimeLevel(msg.getDelayLevel());
		SendResult sendResult = doSend(mqMsg);
		logger.info("RocketMQ消息发送成功: msg={" + msg.toString() + "},topic={" + topic + "},tag={" + tag + "},sendResult={"
					+ sendResult.getSendStatus() + "},msgId={" + sendResult.getMsgId() + "}");
		// 转成SendResult封装对象
		DefaultSendResult qdSendResult = new DefaultSendResult();
		qdSendResult.setMessageQueue(sendResult.getMessageQueue());
		qdSendResult.setMsgId(sendResult.getMsgId());
		qdSendResult.setQueueOffset(sendResult.getQueueOffset());
		qdSendResult.setSendStatus(sendResult.getSendStatus());
		qdSendResult.setMqMsg(mqMsg);
		return qdSendResult;
	}
	
	@Override
	public void sendMessage(String topic, String tag, SendMessage msg,SendCallback sendCallback) {
		if (doCheckTopic(topic) || doCheckTag(tag)) {
			throw new MQException(MQRuntimeFailStatus.MQ_PUBLISH_PERMISSION_FAILURE);
		}
		msg.setInstanceName(instanceName);
		byte[] body = null;
		try {
			body = JdkSerializer.serialize(msg);
		} catch (Exception e) {
			logger.error("序列化SendMessage信息异常", e);
		}
		Message mqMsg = new Message(topic, tag, body);
		if (msg.getDelayLevel() > 0) mqMsg.setDelayTimeLevel(msg.getDelayLevel());
		doSendAsync(mqMsg,sendCallback);
		logger.info("RocketMQ消息异步发送成功: msg={" + msg.toString() + "},topic={" + topic + "},tag={" + tag + "}");
	}
	
	private SendResult doSend(Message message) {
		SendResult sendResult = null;
		try {
			sendResult = producer.send(message);
		} catch (MQClientException e) {
			logger.error("RocketMQ发送失败, | " + message.toString(), e);
			throw new MQException(MQRuntimeFailStatus.MQ_CLIENT_FAILURE, e);
		} catch (RemotingException e) {
			logger.error("RocketMQ发送失败, | " + message.toString(), e);
			throw new MQException(MQRuntimeFailStatus.MQ_REMOTING_FAILURE, e);
		} catch (MQBrokerException e) {
			logger.error("RocketMQ发送失败, | " + message.toString(), e);
			throw new MQException(MQRuntimeFailStatus.MQ_BROKER_FAILURE, e);
		} catch (InterruptedException e) {
			logger.error("RocketMQ发送失败, | " + message.toString(), e);
			throw new MQException(MQRuntimeFailStatus.MQ_INTERRUPT_FAILURE, e);
		}
		return sendResult;
	}
	
	private void doSendAsync(Message message,SendCallback sendCallback) {
		try {
			producer.send(message, sendCallback);
		} catch (MQClientException e) {
			logger.error("RocketMQ发送失败, | " + message.toString(), e);
			throw new MQException(MQRuntimeFailStatus.MQ_CLIENT_FAILURE, e);
		} catch (RemotingException e) {
			logger.error("RocketMQ发送失败, | " + message.toString(), e);
			throw new MQException(MQRuntimeFailStatus.MQ_REMOTING_FAILURE, e);
		} catch (InterruptedException e) {
			logger.error("RocketMQ发送失败, | " + message.toString(), e);
			throw new MQException(MQRuntimeFailStatus.MQ_INTERRUPT_FAILURE, e);
		}
	}
	
	public void checkPropertiesAndSetEnv() {
		if (StringUtils.isEmpty(namesrvAddr) || StringUtils.isEmpty(instanceName)) {
			logger.error("请设置namesrvNo 和 instanceName" );
			throw new MQException(MQRuntimeFailStatus.MQ_INIT_PARAM_FAILURE);
		}
		if (StringUtils.isEmpty(System.getProperty("rocketmq.namesrv.addr"))) {
			System.setProperty("rocketmq.namesrv.addr", namesrvAddr);
		}
		if (StringUtils.isEmpty(System.getProperty("rocketmq.publisher.name"))) {
			System.setProperty("rocketmq.publisher.name", instanceName);
		}
		// 如果在配置中没有设置allowedTopics属性,则认为对可以发送的topics不做检查
		if (StringUtils.isEmpty(topicsACL)) {
			needCheckTopics = false;
		}
		// 如果在配置中没有设置allowedTags属性,则认为对可以发送的tags不做检查
		if (StringUtils.isEmpty(tagsACL)) {
			needCheckTags = false;
		}
	}
	
	public boolean doCheckTopic(String topic) {
		if (needCheckTopics && StringUtils.isNotEmpty(topicsACL)) {
			if (topicsACL.contains(topic)) {
				return true;
			}
			logger.error("This topic is not in allowed topic list:topic={" + topic + "},allowed topic list={" + topicsACL + "}");
			return false;
		}
		return false;
	}
	
	public boolean doCheckTag(String tag) {
		if (needCheckTags && StringUtils.isNotEmpty(tagsACL)) {
			if (topicsACL.contains(tag)) {
				return true;
			}
			logger.error("This tag is not in allowed tag list:tag={" + tag + "},allowed tag list={" + tagsACL + "}");
		}
		return false;
	}
	
	public void destroy() {
		if (producer != null) {
			producer.shutdown();
			logger.info("RocketMQ 发送端关闭成功;rocketmq.namesrv.addr={" + namesrvAddr + "}");
		}
	}
	
	public String getNamesrvAddr() {
		return namesrvAddr;
	}
	
	public void setNamesrvAddr(String namesrvAddr) {
		this.namesrvAddr = namesrvAddr;
	}
	
	public String getInstanceName() {
		return instanceName;
	}
	
	public void setInstanceName(String instanceName) {
		this.instanceName = instanceName;
	}
	
	public String getTopicsACL() {
		return topicsACL;
	}
	
	public void setTopicsACL(String topicsACL) {
		this.topicsACL = topicsACL;
	}
	
	public String getTagsACL() {
		return tagsACL;
	}
	
	public void setTagsACL(String tagsACL) {
		this.tagsACL = tagsACL;
	}
	
	public String getGroup() {
		return group;
	}
	
	public void setGroup(String group) {
		this.group = group;
	}
	
	public int getRetryTimesWhenSendFailed() {
		return retryTimesWhenSendFailed;
	}
	
	public void setRetryTimesWhenSendFailed(int retryTimesWhenSendFailed) {
		this.retryTimesWhenSendFailed = retryTimesWhenSendFailed;
	}
}
